package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.util


object Demo18Code {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("code")
    val sc = new SparkContext(conf)

    //读取分数表
    val scoresRDD: RDD[(String, String, Double)] = sc
      .textFile("data/score.txt")
      .map(_.split(","))
      .map { case Array(id, cid, score) => (id, cid, score.toDouble) }

    //读取学生表
    val studentRDD: RDD[(String, String, Int, String, String)] = sc
      .textFile("data/students.txt")
      .map(_.split(","))
      .map { case Array(id, name, age, sex, clazz) => (id, name, age.toInt, sex, clazz) }

    //读取科目表
    val subjectRDD: RDD[(String, String, String)] = sc
      .textFile("data/subject.txt")
      .map(_.split(","))
      .map { case Array(cid, cname, maxScore) => (cid, cname, maxScore) }

    //1,统计每个班级总分前十的学生
    //计算总分
    val sumScoreRDD: RDD[(String, Double)] = scoresRDD
      .map { case (id, _, score) => (id, score) }
      .reduceByKey(_ + _)


    val idAndClazzRDD: RDD[(String, String)] = studentRDD
      .map { case (id, _, _, _, clazz) => (id, clazz) }

    //关联学生表获取班级
    val idClazzScore: RDD[(String, String, Double)] = sumScoreRDD
      .join(idAndClazzRDD)
      .map { case (id, (score, clazz)) => (id, clazz, score) }

    //按照班级分组
    val clazzGroupByRDD: RDD[(String, Iterable[(String, String, Double)])] = idClazzScore
      .groupBy { case (id, clazz, score) => clazz }

    //组内取前十
    clazzGroupByRDD.map {
      case (clazz, iter) =>
        //组内的数据可能会很多，如果转换成列表会导致内存溢出
        //iter.toList.sortBy(-_._3).take(10)
        //定义链表保存前十的数据
        val top10 = new util.LinkedList[(String, String, Double)]()
        iter.foreach {
          case (id, clazz, score) =>
            var index = 0
            var flag = true
            //循环比较结果中每一个元素和新的元素的大小i
            while (index < top10.size() && flag) {
              val item: (String, String, Double) = top10.get(index)
              //如果新的元素大于当前元素，将新的元素插入当前元素的前面
              if (score > item._3) {
                top10.add(index, (id, clazz, score))
                flag = false
              }
              index += 1
            }
            //如果新的元素小于解雇欧中的所有元素，插入后面
            if (flag) {
              top10.add((id, clazz, score))
            }
            //如果结果超过10个，删除最后一个
            if (top10.size() == 11) {
              top10.removeLast()
            }
        }
        top10
    }.foreach(println)

    while (true) {}
  }
}